package com.aotain.nyx.bots;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.statis.AbnStatisTuple;

public class BotsFlink implements FlatMapFunction<Tuple2<String,BotsTuple>, AbnStatisTuple> {
	
	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = -4037046728657709277L;

	/**
	 * 固定源60s内握手成功次数  目标|源|端口
	 */
	private HashMap<String,Integer> fixedSourceMap = new HashMap<String,Integer>();

	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> upStreamoctets = new HashMap<String,Long>();

	/**
	 * 上行包数
	 */
	private HashMap<String,Long> upStreampacket = new HashMap<String,Long>();

	/**
	 * 攻击开始时间
	 */
	private HashMap<String,Long> attackStartMap = new HashMap<String,Long>();

	/**
	 * 攻击持续的上次更新时间
	 */
	private HashMap<String,Long> attackLasttimeMap = new HashMap<String,Long>();

	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;

	@Override
	public void flatMap(Tuple2<String, BotsTuple> value,
			Collector<AbnStatisTuple> out) throws Exception {
		// TODO Auto-generated method stub
		if(value.f0 == null) {
			return;
		}
		
		upStreamoctets.put(value.f0, value.f1.getUpStreamOctets());
		upStreampacket.put(value.f0, value.f1.getUpStreamPacket());
		
		try {
			//		InsertIntoData();
			for(Entry<String, Long> entry : upStreamoctets.entrySet()) {

				String entryKey = entry.getKey();
				long currenttime = System.currentTimeMillis();
	
				String[] tuplesp = value.f0.split("\\|",-1);
				String dip = tuplesp[0];
				String sip = tuplesp[1];
				
				if(upStreamoctets.get(entryKey)/(1024*1024*60) > 50 && fixedSourceMap.get(entryKey) > 18000) {//如果每秒流量大于50M,连接数大于18000则判断有攻击

					if(attackLasttimeMap.containsKey(entryKey)) {//如果存在就更新

						attackLasttimeMap.put(entryKey, currenttime);
						UpdateData(entry,out);
					}else {//不存在就新入数据
						attackStartMap.put(entryKey, currenttime);
						attackLasttimeMap.put(entryKey, currenttime);
						InsertIntoData(entry,currenttime,out);
					}
				}else {
					//如果没大于门限 也时间超过3分钟 则攻击结束
					if (attackLasttimeMap.containsKey(entryKey)) {
						if(currenttime - attackLasttimeMap.get(entryKey) > END_DURATION) {

							EndIntoData(entryKey);
							attackLasttimeMap.remove(entryKey);
							attackStartMap.remove(entryKey);
						}
					}
				}
			}

			// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束，避免漏处理数据
			ArrayList<String> removeItem = new ArrayList<String>();
			for(Entry<String,Long> entry:attackLasttimeMap.entrySet()) {
				String entryKey = entry.getKey();
				if(System.currentTimeMillis() - entry.getValue() > END_DURATION) {
					removeItem.add(entryKey);
					EndIntoData(entryKey);
					attackStartMap.remove(entryKey);
				}
			}

			for(String key  : removeItem) {
				attackLasttimeMap.remove(key);
			}

			removeItem.clear();
			upStreamoctets.clear();
			upStreampacket.clear();
			fixedSourceMap.clear();
		} catch (Exception e ) {
			e.printStackTrace();
			Logger.getLogger(getClass()).error("SCANATTACK Exception",e);;
		}
	}
	
	/**
	 * 新记录入库
	 * @Title: InsertIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	//	public void InsertIntoData() {
	public void InsertIntoData(Entry<String, Long> entry,long time,Collector<AbnStatisTuple> out) {


		String entryKey = entry.getKey();
		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String dport = tuplesp[2];
		String gis = tuplesp[3];
		String idc = tuplesp[4];
		String flowDir = tuplesp[5];

		String[] gisArray = gis.split("#");								
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];		
		String sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(time));
		String rowKeyAbnormal = accesstime + "_" + sip + "_"  + dip + "_" + dport + "_15";	
	 	
		String desc = String.format("僵尸主机,连接次数[%d],攻击峰值流量[%f KB/s]", fixedSourceMap.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));
		System.out.println("BotsFlink----InsertIntoData" + rowKeyAbnormal  + " : " + desc);
		
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("ROWKEY",rowKeyAbnormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP",sip));
		row.add(new DataColumn("DESTPORT",dport));
		row.add(new DataColumn("ACCESSTIME",Long.parseLong(accesstime)));
		row.add(new DataColumn("ABRNORMAL", 15));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", upStreampacket.get(entryKey).toString()));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("PROVINCE", sourceProvinceName));
		row.add(new DataColumn("FLOWDIRECTION", flowDir));
		row.add(new DataColumn("UPSTREAMOCTETS", upStreamoctets.get(entryKey)));
		row.add(new DataColumn("UPSTREAMPACKET", upStreampacket.get(entryKey)));
		row.add(new DataColumn("ATTTYPE","2"));


		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("BOTS");
		record.setUpStreamOctets(upStreamoctets.get(entryKey));
		record.setUpStreamPacket(upStreampacket.get(entryKey));
		record.setAttackNum(1l);
		record.setSourceCountry(sourceAreaCountry);
		record.setSourceProvince(sourceProvinceName);
		record.setAttackType("2");
		record.setAttTime(accesstime);
		out.collect(record);
		
		Logger.getLogger(BotsFlink.class).info(
				String.format("###BOTS V2 ATTACK START ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, entryKey));

	}

	/**
	 * 更新记录字段
	 * @Title: UpdateData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void UpdateData(Entry<String, Long> entry,Collector<AbnStatisTuple> out) {


		
		String entryKey = entry.getKey();
		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String dport = tuplesp[2];
		String gis = tuplesp[3];
		String[] gisArray = gis.split("#");								
		String sourceAreaCountry = gisArray[4];
		String sourceProvinceName =  gisArray[6].trim();
		
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		String rowKeyAbnormal = accesstime + "_" + sip + "_"  + dip + "_" + dport + "_15";
		String desc = String.format("僵尸主机 ,连接次数[%d],攻击峰值流量[%f KB/s]", fixedSourceMap.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));

		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("UPSTREAMOCTETS", upStreamoctets.get(entryKey)));
		row.add(new DataColumn("UPSTREAMPACKET", upStreampacket.get(entryKey)));
		row.add(new DataColumn("ATTNUM", upStreampacket.get(entryKey).toString()));
		row.add(new DataColumn("DESC", desc));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);

		Logger.getLogger(BotsFlink.class).info(
				String.format("###BOTS V2 ATTACK UPDATE ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, entryKey));
		
		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("BOTS");
		record.setUpStreamOctets(upStreamoctets.get(entryKey));
		record.setUpStreamPacket(upStreampacket.get(entryKey));
		record.setAttackNum(0l);
		record.setSourceCountry(sourceAreaCountry);
		record.setSourceProvince(sourceProvinceName);
		record.setAttackType("2");
		record.setAttTime(accesstime);
		out.collect(record);
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);

		//		String rowkey3 = String.format("%s_%s", accesstime, sip);
		//		hbaseInstance.Add(ApolloProperties.SDS_ABNORMAL_LOG3, rowkey3, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
	}

	/**
	 * 攻击结束记录入库
	 * @Title: EndIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param entryKey
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void EndIntoData(String entryKey) {

		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String dport = tuplesp[2];
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));

		String rowKeyAbnormal = accesstime + "_" + sip + "_"  + dip + "_" + dport + "_15";
		String desc = String.format("僵尸主机结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)+60*1000l-attackStartMap.get(entryKey))/1000);
		
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("ENDTIME", Long.parseLong(new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackLasttimeMap.get(entryKey)+60*1000l)))));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);

		Logger.getLogger(BotsFlink.class).info(
				String.format("###BOTS V2 ATTACK END ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, entryKey));

		
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ENDTIME", new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackLasttimeMap.get(entryKey)+60*1000l)));

	}
		

}
